/*
 * Copyright 2011-2013, David George, Licensed under the Apache License,
 * Version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package org.magneato.dao;

import java.io.File;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.magneato.dto.KVResults;
import org.magneato.dto.Page;
import org.magneato.service.ExportHelper;
import org.magneato.service.PKHelper;
import org.magneato.utils.Conversion;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.datasource.lookup.DataSourceLookupFailureException;
import org.springframework.stereotype.Repository;

import fx.sunjoy.algo.impl.DiskTreap;
import fx.sunjoy.algo.impl.DiskTreapNode;
import fx.sunjoy.utils.FastString;

/**
 * Treapdb implementation of the Key/Value store interface
 * 
 * @author David B George
 * @see http://
 */
@SuppressWarnings("rawtypes")
@Repository
public class TreapStore<K> implements KVStore {
	private DiskTreap<FastString, byte[]> treap;
	MessageDigest md;
	private final Map<String, Lock> commentLocks = new HashMap<String, Lock>();

	private PKHelper pkHelper;
	private int places;
	private int max;
	private final static int SHALen = 20;
	private final static char SEP = ':';

	private static final Logger _logger = LoggerFactory
			.getLogger(TreapStore.class);

	public void init(String treapDir, int maxRelated, PKHelper p) {
		File indexFile = new File(treapDir);

		try {
			md = MessageDigest.getInstance("SHA-1");
		} catch (NoSuchAlgorithmException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}

		// keylen is SHALen + 2 + 5 + 5 = 32 /e.g.
		// PPPPPPPPPPPPPPPPPPPIICCCCCen_US
		int ibz = 32 + DiskTreapNode.STRING_KEY_OVER_HEAD;
		try {
			treap = new DiskTreap<FastString, byte[]>(ibz, indexFile);
		} catch (Exception e) {
			String errorMsg = e.getMessage();
			_logger.error(errorMsg);
			throw new DataSourceLookupFailureException(errorMsg);
		}
		_logger.info("Store " + treapDir + " currently has " + treap.length()
				+ " objects");

		this.pkHelper = p;
		places = Integer.toString(maxRelated).length();
		max = maxRelated;
	}

	public int length() {
		return treap.length();
	}

	@Override
	/**
	 * Check if SEP - comment
	 * Split out name part and add 00, if object name same as key then return, otherwise increment index and continue.
	 */
	public Object get(String key) {
		byte[] commentNo = {};
		byte[] pk;

		if (key == null) {
			throw new IllegalArgumentException("Key cannot be null");
		}

		try {
			int i = key.indexOf(SEP);
			if (i > 0) {
				// related content, split the key from the related content
				// number
				String index = key.substring(i + 1);
				commentNo = StringUtils.leftPad(index, places, '0').getBytes();

				key = key.substring(0, i);
			}
			/*
			 * Transform a variable length string to a 20 byte key
			 */
			pk = md.digest(key.getBytes("UTF-8"));

			pk = Conversion.concat(pk, commentNo);
			byte[] bytes = treap.get(new FastString(pk));
			return Conversion.Bytes2Object(bytes);
		} catch (UnsupportedEncodingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * Key Format keylen is SHALen + 2 + 5 + 5 = 32 /e.g.
	 * PPPPPPPPPPPPPPPPPPPIICCCCCen_US
	 * 
	 * @key Primary key, can be followed by a SEP which means a related page. eg.
	 *      page or page:1
	 */
	@Override
	public void put(final String key, Object value) {
		byte[] commentNo = {};
		byte[] pk;

		if (key == null) {
			throw new IllegalArgumentException("Key cannot be null");
		}
		String cKey = key;

		try {
			int i = key.indexOf(SEP);
			if (i > 0) {
				// related content, split the key from the related content
				// number
				String index = key.substring(i + 1);
				commentNo = StringUtils.leftPad(index, places, '0').getBytes();

				cKey = key.substring(0, i);
			}

			pk = md.digest(cKey.getBytes("UTF-8"));
			pk = Conversion.concat(pk, commentNo);

			if (checkClash(pk, key)) {
				// TODO: change this
				throw new RuntimeException("Key clash");
			}

			/**
			 * We need to take account of key clashes. These are extremely rare.
			 * We tested the Wikipedia index with SHA and found no clashes.
			 * 
			 * to do this we fetch the value and check the page name, if there
			 * is a class we modify the key.
			 */

			treap.put(new FastString(pk), Conversion.Object2Bytes(value));
		} catch (IOException e) {
			// TODO Auto-generated catch block
			_logger.error(e.getLocalizedMessage());
		}
	}

	/*
	 * Returns index so that key + index is uniq
	 */
	private boolean checkClash(byte[] key, String name) {

		FastString k1 = new FastString(key);
		byte[] bytes = treap.get(k1);
		if (bytes != null) {
			Object o = Conversion.Bytes2Object(bytes);

			if (o != null) {
				String k2 = pkHelper.getPK(o);
				if (!name.equals(k2)) {
					_logger.error("Key clash on " + name + " and " + k2);
					return true;
				}
			}
		}
		return false;
	}

	/*
	 * Deletes all related pages (comments) or if a comment is specified, a
	 * single comment
	 * 
	 * @see org.magneato.dao.KVStore#delete(java.lang.String)
	 */
	@Override
	public String delete(final String key) {
		byte[] bytesOfMessage;

		try {
			int i = key.indexOf(SEP);
			if (i > 0) {
				// related
				String index = key.substring(i + 1);
				String stem = key.substring(0, i);
				byte[] hash = md.digest(stem.getBytes("UTF-8"));
				byte[] commentNo = StringUtils.leftPad(index, places, '0')
						.getBytes();
				hash = Conversion.concat(hash, commentNo);
				treap.delete(new FastString(hash));
				return stem;
			} else {
				bytesOfMessage = key.getBytes("UTF-8");
				byte[] hash = md.digest(bytesOfMessage);

				treap.removePrefix(new FastString(hash));
				return key;
			}
		} catch (UnsupportedEncodingException e) {
			_logger.error("Delete " + e.getLocalizedMessage());
		}
		return null;
	}

	public void export(ExportHelper p) {
		int length = treap.length();

		for (int i = 1; i < length + 1; i++) {
			Map.Entry<FastString, byte[]> mapEntry = treap.kth(i, true);
			byte[] bytes = mapEntry.getValue();
			Object o = Conversion.Bytes2Object(bytes);
			p.toXML(o);
		}
	}

	/**
	 * Test method to print name of all objects in store
	 */
	public void titles(int s, int len) {
		int length = treap.length();

		for (int i = s; i < s + len && i < length + 1; i++) {
			Map.Entry<FastString, byte[]> mapEntry = treap.kth(i, true);
			if (mapEntry != null) {
				byte[] bytes = mapEntry.getValue();
				Object o = Conversion.Bytes2Object(bytes);
				Page p = (Page) o;
			}
		}
	}

	/******************** Related pages - used for comments *********************/
	@Override
	public int addRelated(String key, Object value) {
		byte[] bytesOfMessage;
		int n = -1;
		Lock lock = null;

		// acquire lock on key
		synchronized (commentLocks) {
			lock = commentLocks.get(key);

			if (lock == null) {
				lock = new Lock();
				commentLocks.put(key, lock);
			} else {
				lock.refCount++;
			}
		}

		try {
			bytesOfMessage = key.getBytes("UTF-8");
			byte[] hash = md.digest(bytesOfMessage);
			// we need to lock here to avoid two threads accessing the last
			// comment number for the key at the same time.
			// First comment is always :1 not :0

			synchronized (lock) {
				n = getLastCommentNumber(key);
				n++;
				byte[] commentNo = String.format("%0" + places + "d", n)
						.getBytes();
				byte[] K = Conversion.concat(hash, commentNo);
				pkHelper.setPK(value, "" + SEP + n);
				treap.put(new FastString(K), Conversion.Object2Bytes(value));
			}
		} catch (UnsupportedEncodingException e) {
			// Log and throw
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			// release lock
			synchronized (commentLocks) {
				if (lock.refCount == 0) {
					commentLocks.remove(key);
				} else {
					lock.refCount--;
				}
			}
		}

		return n;
	}

	/**
	 * Note: This isn't hugely fast but it is only called on adds
	 * 
	 * TODO: assumes 5 digit comment id length and 20 digit hash length.
	 * 
	 * @param key
	 * @return
	 * @throws UnsupportedEncodingException
	 */
	private int getLastCommentNumber(String key) {
		byte[] bytesOfMessage = null;
		try {
			bytesOfMessage = key.getBytes("UTF-8");
		} catch (UnsupportedEncodingException e) {
			_logger.error(e.getLocalizedMessage());
			throw new RuntimeException(e);
		}
		byte[] K = Conversion.concat(md.digest(bytesOfMessage));

		Map<FastString, byte[]> map = treap.prefix(new FastString(K), 1, null,
				false);
		if (map.isEmpty()) {
			return 0;
		}

		FastString fs = map.keySet().iterator().next();
		if (fs.bytes.length == SHALen) {
			return 0;
		}

		StringBuilder sb = new StringBuilder();
		for (int c = SHALen; c < (SHALen + places); c++) {
			sb.append(String.format("%c", fs.bytes[c]));
		}
		return Integer.parseInt(sb.toString());
	}

	@Override
	public KVResults getRelated(String key, int page, int count) {
		byte[] bytesOfMessage;
		try {
			bytesOfMessage = key.getBytes("UTF-8");
			byte[] hash = md.digest(bytesOfMessage);

			int start = (page * count) + 1;

			List<Object> list = new ArrayList<Object>(count);
			Map<FastString, byte[]> map = treap
					.range(new FastString(Conversion.concat(hash, String
							.format("%0" + places + "d", start).getBytes())),
							new FastString(Conversion.concat(hash, String
									.format("%0" + places + "d", max)
									.getBytes())), count);

			for (byte[] b : map.values()) {
				Object o = Conversion.Bytes2Object(b);
				list.add(o);
			}
			KVResults relatedObjects = new KVResults<Object>(page, count,
					getLastCommentNumber(key), list);

			return relatedObjects;

		} catch (UnsupportedEncodingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		return null;
	}

	@Override
	public void close() {
		treap.close();
	}

	private class Lock {
		@SuppressWarnings("unused")
		public int refCount = 0;
	}

	@Override
	public void deleteRelated(String key) {
		// TODO Auto-generated method stub

	}
}
